package com.hex.ds.hdrs.engine.service;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.hex.ds.hdrs.app.service.AppManager;
import com.hex.ds.hdrs.app.service.AppSession;
import com.hex.ds.hdrs.center.service.CenterManager;
import com.hex.ds.hdrs.common.constant.ErrorCode;
import com.hex.ds.hdrs.common.exception.BizException;
import com.hex.ds.hdrs.common.utils.ListUtil;
import com.hex.ds.hdrs.engine.exec.ProcessorExec;
import com.hex.ds.hdrs.engine.manager.LstFinishManager;
import com.hex.ds.hdrs.engine.manager.RunningManager;
import com.hex.ds.hdrs.engine.manager.ToDoManager;
import com.hex.ds.hdrs.engine.manager.ToRunManager;
import com.hex.ds.hdrs.label.service.ILabelJobService;
import com.hex.ds.hdrs.period.job.po.JobTask;
import com.hex.ds.hdrs.period.job.po.PeriodJob;
import com.hex.ds.hdrs.period.job.po.PeriodJobDef;
import com.hex.ds.hdrs.period.job.service.PeriodJobDefService;
import com.hex.ds.hdrs.period.job.service.PeriodJobService;
import com.hex.ds.hdrs.period.task.po.PeriodTask;
import com.hex.ds.hdrs.period.task.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;

/**
 * @Package: com.hex.ds.hdrs.engine.service
 * @ClassName ProcessorService
 * @Description: 处理器
 * @Author: jd.chen
 * @Date 2021/7/5 18:46
 * @Version v1.0
 **/
@Slf4j
@Service
public class ProcessorService {

    private static List<ProcessorExec> processorThreadList = new ArrayList();

    // 依赖检查走db
    private static final String DEAL_TYPE_DB = "db";
    // 存在上游依赖未完成
    private static final String DEAL_TYPE_NO = "no";
    // 上游依赖全部已完成，任务可执行
    private static final String DEAL_TYPE_YES = "yes";

    @Autowired
    private CenterManager centerManager;
    @Autowired
    private AppManager appManager;
    @Autowired
    private TaskService taskService;
    @Autowired
    private ToDoManager toDoManager;
    @Autowired
    private ToRunManager toRunManager;
    @Autowired
    private RunningManager runningManager;
    @Autowired
    private LstFinishManager lstFinishManager;
    @Autowired
    private ILabelJobService iLabelJobService;
    @Autowired
    private PeriodJobService periodJobService;
    @Autowired
    private PeriodJobDefService periodJobDefService;

    public ToDoManager getToDoManager() {
        return toDoManager;
    }

    public ToRunManager getToRunManager() {
        return toRunManager;
    }

    public CenterManager getCenterManager() {
        return centerManager;
    }

    public ILabelJobService getILabelJobService() {
        return iLabelJobService;
    }

    public PeriodJobService getPeriodJobService() {
        return periodJobService;
    }

    public List<ProcessorExec> getThreadList() {
        return processorThreadList;
    }

    /**
     * 初始化
     */
    public void init() {
        log.debug("处理器--初始化【ProcessorService.init】开始=======》");

        if (!ListUtil.isEmpty(processorThreadList)) {
            log.warn("【警告】【ProcessorService.init】处理器线程集合不为空，清空线程集合………………");
            processorThreadList.clear();
        }

        try {
            // 创建线程池
            ProcessorExec processorExec;
            for (int i = 0; i < toDoManager.getPoolSize(); i++) {
                processorExec = new ProcessorExec(this, "processor-thread-" + i);
                processorThreadList.add(processorExec);
                processorExec.start();
            }
        } catch (Exception e) {
            log.error("处理器--初始化【ProcessorService.run】异常: {}", e.getMessage());

            processorThreadList.clear();
            throw new BizException(ErrorCode.ERROR_810003);
        } finally {
            log.debug("处理器--初始化【ProcessorService.init】结束《=======");
        }
    }

    /**
     * 从queue获取作业任务
     *
     * @return
     */
    public JobTask getJobTaskFromQueue() {
        JobTask jobTask = toDoManager.poll();
        log.debug("从queue获取作业任务【getJobTaskFromQueue】");

        if (jobTask == null) {
            return null;
        }

        if (this.delJob(jobTask)) {
            return null;
        }

        if (!appManager.isValidAppSession(jobTask.getAppCode())) {
            log.warn("【ProcessorService.getJobTaskFromQueue】该任务所属的应用未启用======");
            toDoManager.addJobToQueue(jobTask);
            return null;
        }

        return jobTask;
    }

    /**
     * 作业是否在delmap，若在，需要删除
     *
     * @param jobTask
     * @return
     */
    private boolean delJob(JobTask jobTask) {
        log.debug("作业是否在delmap，若在，需要删除【ProcessorService.delJob】=======》");

        if (toDoManager.isContainsDelJob(jobTask.getJobCode())) {
            log.warn("该作业/任务在deltaDelJobMap中【ProcessorService.delJob】job【{}】【{}】", jobTask.getJobCode(), jobTask.getBizDate());
            toDoManager.removeDelMapJob(jobTask);
            return true;
        }

        return false;
    }

    /*
     * @Method: checkRunTime <br>
     * @Param: [task] <br>
     * @Return: void <br>
     * @Description：判断任务是否到达执行时间<br>
     * @Author： wz.li<br>
     * @Date： 2022/12/10 17:13 <br>
     * @Version： V2.0.2<br>
     */
    public void checkRunTime(PeriodTask periodTask) {
        Date runningTime = DateUtil.date(DateUtil.parse(periodTask.getBizDate()));
        runningTime = DateUtil.offsetHour(runningTime, getOffsetHour(periodTask));
        if (DateUtil.date().before(runningTime)) {
            throw new BizException(ErrorCode.ERROR_821006);
        }
    }

    private Integer getOffsetHour(PeriodTask periodTask){
        if (periodTask.getOffsetHour() == null){
            PeriodJobDef periodJobDef = periodJobDefService.getJobDefByJobCode(periodTask.getJobCode());
            periodTask.setOffsetHour(periodJobDef.getOffsetHour());
        }
        return periodTask.getOffsetHour();
    }

    /**
     * 检查互斥作业/任务是否有在执行中
     *
     * @param task
     * @return
     */
    public void checkMutex(PeriodTask task) {
        String taskCode = task.getTaskCode();
        log.debug("检查互斥作业/任务是否有在执行中【任务编码: {}】", taskCode);
        AppSession appSession = appManager.getValidAppSession(task.getAppCode());

        List<String> mutexTaskDefList = appSession.getMutexTaskCodeList();
        if (ListUtil.isEmpty(mutexTaskDefList)) {
            log.debug("该应用下无互斥任务【任务编码: {}】", taskCode);
            return;
        }
        if (mutexTaskDefList.contains(taskCode)) {
            for (String taskDefCode : mutexTaskDefList) {
                if (runningManager.isContainsTask(taskDefCode)) {
                    log.warn("任务【任务编码: {}】不允许执行,该任务的互斥任务【任务编码: {}】正在执行", taskCode, taskDefCode);
                    String errorMsg = StrUtil.format(ErrorCode.ERROR_821000.getMessage(), taskDefCode);
                    throw new BizException(ErrorCode.ERROR_821000,errorMsg);
                }
            }
        }
    }

    /**
     * 检查上游依赖
     *
     * @param task
     * @return true(无依赖) false(有依赖)
     */
    public void checkUpDepend(PeriodTask task) {
        log.debug("【任务上游依赖检查】检查任务任务编码: {},业务日期: {}的上游依赖", task.getTaskCode(), task.getBizDate());
        this.checkUpJobDepend(task);
        this.checkUpTaskDepend(task);
    }

    /**
     * 检查上游作业依赖
     *
     * @param curTask
     * @return
     */
    public void checkUpJobDepend(PeriodTask curTask) {
        String taskCode = curTask.getTaskCode();
        String bizDate = curTask.getBizDate();
        log.debug("【任务上游依赖检查】检查任务编码: {}, 业务日期: {}的上游依赖作业完成情况", taskCode, bizDate);
        // 1. 通过taskCode获取上游依赖
        List<String> upJobList = StrUtil.isBlank(curTask.getDependJob()) ? new ArrayList<>() : Arrays.asList(curTask.getDependJob().split(","));
        if (ListUtil.isEmpty(upJobList)) {
            log.debug("【任务上游依赖检查】检查任务编码: {}, 业务日期: {},没有上游依赖作业", taskCode, bizDate);
            return;
        }
        String errorMsg = taskService.getUpDependErrorMsg(curTask,ErrorCode.ERROR_821004.getMessage());
        if (StrUtil.isNotBlank(errorMsg)){
            log.debug("db校验依赖【任务编码: {}】【{}】上游作业未完成!", taskCode, bizDate);
            throw new BizException(ErrorCode.ERROR_821004,errorMsg);
        }
    }

    /**
     * 检查上游任务依赖
     *
     * @param curTask
     * @return
     */
    public void checkUpTaskDepend(PeriodTask curTask) {
        String taskCode = curTask.getTaskCode();
        String bizDate = curTask.getBizDate();
        log.debug("【checkUpTaskDepend】【{}】【{}】", taskCode, bizDate);

        // 1. 通过taskCode获取上游依赖
        List<String> upTaskDefList = StrUtil.isBlank(curTask.getDependTask()) ? new ArrayList<>() : Arrays.asList(curTask.getDependTask().split(","));
        if (ListUtil.isEmpty(upTaskDefList)) {
            log.debug("【任务编码: {}】没有上游依赖", taskCode);
            return;
        }
        String errorMsg = taskService.getUpDependErrorMsg(curTask,ErrorCode.ERROR_821004.getMessage());
        if (StrUtil.isNotBlank(errorMsg)){
            log.debug("db校验依赖【任务编码: {}】【{}】上游任务未完成！", taskCode, bizDate);
            throw new BizException(ErrorCode.ERROR_821004,errorMsg);
        }
    }

    /**
     * 内存中校验上游作业依赖
     *
     * @param upJobDefList
     * @return
     */
    private String checkJobByMemory(List<String> upJobDefList, String curBizDate) {
        log.debug("内存中校验上游作业依赖【ProcessorService.checkJobByMemory】=======》");
        PeriodJob latestJob;
        for (String upJobCode : upJobDefList) {
            latestJob = lstFinishManager.getJob(upJobCode);
            if (null != latestJob) {
                if (DateUtil.parse(latestJob.getBizDate()).before(DateUtil.parse(curBizDate))) {
                    log.debug("【ProcessorService.checkJobByMemory】上游作业未完成======");
                    // 上游依赖作业业务时间小于该作业业务时间，表明上游依赖未完成
                    return DEAL_TYPE_NO;
                }

                if (DateUtil.parse(latestJob.getBizDate()).after(DateUtil.parse(curBizDate))) {
                    log.debug("【ProcessorService.checkJobByMemory】db校验依赖======");
                    // 数据库校验
                    return DEAL_TYPE_DB;
                }
            } else {
                return DEAL_TYPE_DB;
            }
        }
        return DEAL_TYPE_YES;
    }

    /**
     * 内存中校验上游任务依赖
     *
     * @param upTaskDefList
     * @return
     */
    private String checkTaskByMemory(List<String> upTaskDefList, String curBizDate) {
        log.debug("内存中校验上游任务依赖【ProcessorService.checkTaskByMemory】=======》");
        PeriodTask latestTask;
        for (String upTaskCode : upTaskDefList) {
            latestTask = lstFinishManager.getTask(upTaskCode);
            if (null != latestTask) {

                if (DateUtil.parse(latestTask.getBizDate()).before(DateUtil.parse(curBizDate))) {
                    log.debug("【ProcessorService.checkTaskByMemory】上游任务未完成======");
                    // 上游依赖任务业务时间小于该任务业务时间，表明上游依赖未完成
                    return DEAL_TYPE_NO;
                }

                if (DateUtil.parse(latestTask.getBizDate()).after(DateUtil.parse(curBizDate))) {
                    log.debug("【ProcessorService.checkTaskByMemory】db校验依赖======");
                    // 数据库校验
                    return DEAL_TYPE_DB;
                }
            } else {
                return DEAL_TYPE_DB;
            }
        }
        return DEAL_TYPE_YES;
    }

    /**
     * 暂停处理器
     */
    public void stopProcess() {
        log.debug("暂停处理器【ProcessorService.stopProcess】开始===》【{}】", DateUtil.now());
        try {
            if (ListUtil.isEmpty(processorThreadList)) {
                log.error("++++++【stopDispatcher】分发器线程为空++++++");
                return;
            }

            for (ProcessorExec processorExec : processorThreadList) {
                processorExec.stopThread();
            }
            //判断处理器的线程是否都执行完
            boolean flag = true;
            while (flag) {
                flag = false;
                for (ProcessorExec processorExec : processorThreadList) {
                    if (processorExec.getRunningFlag()) {
                        flag = true;
                        break;
                    }
                }
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            log.error("暂停处理器【ProcessorService.stopProcess】异常: {}", e.getMessage());

        } finally {
            log.debug("暂停处理器【ProcessorService.stopProcess】结束《====【{}】", DateUtil.now());
        }
    }

    /**
     *
     */
    public void continueProcess() {
        log.debug("继续处理器【ProcessorService.continueProcess】开始==》");
        for (ProcessorExec processorExec : processorThreadList) {
            processorExec.continueThread();
        }
        log.debug("继续处理器【ProcessorService.continueProcess】结束《==");
    }

    /**
     * 关闭处理器
     */
    public void closeProcess() {
        log.info("关闭处理器【ProcessorService.closeProcess】开始===》【{}】", DateUtil.now());
        try {
            for (ProcessorExec processorExec : processorThreadList) {
                processorExec.closeThread();
            }
            processorThreadList.clear();
        } catch (Exception e) {
            log.error("关闭处理器【ProcessorService.closeProcess】异常: {}", e.getMessage());

        } finally {
            log.info("关闭处理器【ProcessorService.closeProcess】结束《===【{}】", DateUtil.now());
        }
    }
}
